using System.Reactive.Linq;
using System.Reactive.Subjects;
using System.Reactive.Concurrency;
using System.Threading.Channels;
using Microsoft.Extensions.Logging;
using ReactiveChannelMessaging.Core;

namespace ReactiveChannelMessaging.Advanced;

/// <summary>
/// 简化的响应式流处理器 - 演示基本的流操作
/// </summary>
public class SimpleStreamProcessor : IAsyncDisposable
{
    private readonly ILogger<SimpleStreamProcessor> _logger;
    private readonly List<IDisposable> _subscriptions;
    private readonly Channel<DataMessage> _inputChannel;
    private readonly Subject<DataMessage> _processedSubject;
    private readonly BehaviorSubject<int> _queueDepth;

    public SimpleStreamProcessor(ILogger<SimpleStreamProcessor> logger)
    {
        _logger = logger;
        _subscriptions = new List<IDisposable>();
        
        _inputChannel = Channel.CreateBounded<DataMessage>(100);
        _processedSubject = new Subject<DataMessage>();
        _queueDepth = new BehaviorSubject<int>(0);

        SetupProcessing();
    }

    /// <summary>
    /// 已处理消息流
    /// </summary>
    public IObservable<DataMessage> ProcessedMessages => _processedSubject.AsObservable();

    /// <summary>
    /// 队列深度流
    /// </summary>
    public IObservable<int> QueueDepth => _queueDepth.AsObservable();

    /// <summary>
    /// 发送消息进行处理
    /// </summary>
    public async Task<bool> SendAsync(DataMessage message)
    {
        try
        {
            await _inputChannel.Writer.WriteAsync(message);
            return true;
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Failed to send message");
            return false;
        }
    }

    private void SetupProcessing()
    {
        // 基本流处理
        var processingSubscription = Observable
            .Create<DataMessage>(observer =>
            {
                var task = Task.Run(async () =>
                {
                    await foreach (var message in _inputChannel.Reader.ReadAllAsync())
                    {
                        try
                        {
                            // 模拟处理
                            await Task.Delay(Random.Shared.Next(10, 100));
                            
                            var processed = message with 
                            { 
                                Content = $"Processed: {message.Content}",
                                Timestamp = DateTime.UtcNow
                            };
                            
                            observer.OnNext(processed);
                            _processedSubject.OnNext(processed);
                            
                            _logger.LogDebug("Message processed: {MessageId}", message.Id);
                        }
                        catch (Exception ex)
                        {
                            _logger.LogError(ex, "Error processing message: {MessageId}", message.Id);
                        }
                    }
                    observer.OnCompleted();
                });
                
                return () => { };
            })
            .Subscribe();

        _subscriptions.Add(processingSubscription);

        // 队列深度监控
        var depthMonitor = Observable
            .Interval(TimeSpan.FromSeconds(1))
            .Subscribe(_ =>
            {
                var depth = Random.Shared.Next(0, 20); // 简化的深度估算
                _queueDepth.OnNext(depth);
            });

        _subscriptions.Add(depthMonitor);
    }

    public async ValueTask DisposeAsync()
    {
        _inputChannel.Writer.TryComplete();
        
        foreach (var subscription in _subscriptions)
        {
            subscription.Dispose();
        }
        
        _processedSubject.Dispose();
        _queueDepth.Dispose();
        
        await Task.CompletedTask;
    }
}